package com.liu.kafka.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.Optional;

/**
 * 单消费
 */
//@Component
@Slf4j
public class SingleConsumer {

    /**
     * 不指定group，默认取yml里配置的
     * errorHandler异常处理监听
     * 不管业务处理是否处理异常还是正常都提交offset 消费异常都统一放到死信队列
     */
    @KafkaListener(topics = {"first"}, groupId = "kafka", errorHandler = "listenerErrorHandler")
    public void onMessage(ConsumerRecord<String, String> consumerRecord, Acknowledgment ack) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (optional.isPresent()) {
            //键
            String key = consumerRecord.key();
            //值
            String value = consumerRecord.value();
            //分区
            int partition = consumerRecord.partition();
            //offset
            long offset = consumerRecord.offset();
            //时间戳
            long timestamp = consumerRecord.timestamp();
            //主题
            String topic = consumerRecord.topic();

            //业务处理
            log.info("----------->  partition:{}  key:{}  value:{} offset:{} ", partition, key, value, offset);

            String name = Thread.currentThread().getName();
            log.info("当前线程名：{}",name);


            try {
                if ("error".equals(value)) {
                    throw new RuntimeException("出现异常了！！！");
                }
            } finally {
                //手动提交offset
                ack.acknowledge();
            }

        }
    }




    /**
     * 手动提交offset  消费者组多消费者
     * 处理异常
     * @param consumerRecord
     * @param ack
     */
//    @KafkaListener(topics = {"TXHKP_SBTZ_MES"}, groupId = "kafka")
    public void MultiConsumer(ConsumerRecord<?, ?> consumerRecord, Acknowledgment ack) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());

        if (optional.isPresent()) {
            try {
                int partition = consumerRecord.partition();
                Object msg = optional.get();
                //业务处理
                log.info("group1-1  partition:{}  message:{}", partition, msg);
                // 手动提交offset
                ack.acknowledge();

            } catch (Exception e) {
                e.printStackTrace();
                //异常消费保存记录
                ack.nack(10);
            }
        }
    }

//    @KafkaListener(topics = {"TXHKP_SBTZ_MES"}, groupId = "kafka")
    public void MultiConsumer2(ConsumerRecord<?, ?> consumerRecord) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());

        int partition = consumerRecord.partition();
        if (optional.isPresent()) {
            Object msg = optional.get();
            log.info("group1-2  partition:{}  message:{}", partition, msg);
        }
    }

    //单消费组
//    @KafkaListener(topics = {"TXHKP_SBTZ_MES"}, groupId = "kafka_test")
    public void SingleConsumer(ConsumerRecord<?, ?> consumerRecord) {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (optional.isPresent()) {
            Object msg = optional.get();
            log.info("group2 message:{}", msg);
        }
    }
}
